package udf.matching;

import java.io.IOException;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;

import org.apache.pig.EvalFunc;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;

import udf.string.Permutations;

public class MultiWordFeatureGenerator extends EvalFunc<DataBag> {

	
	private int _data = 0;
	private int _user=1;
	private int _url = 2;
	private int _tag=3;
	
	//hyphen dictionary
	
	private int _h=0;
	private int _freq_h=1;
	
	
	static private boolean initialized = false;
	private int bag_entry_size = -1;
	private BagFactory bagFactory = BagFactory.getInstance();

	
	//private Hashtable<String,Integer> hyphens = new Hashtable<String, Integer>();

	static private Hashtable<String,LinkedList<String>> ordered = new Hashtable<String, LinkedList<String>>();

	
	// static private final String domain_path =
	// "/user/duartes/helpers/domains.txt";
/**
 * Tuple has the following fields
 * 
 * 
 * 0  -   bag of tuples(bookmark)
 * 
 * 1  -  dictionary to do the look up (hyphenized enriched)
 * 

 * 
 * 
 */
	public void init(Tuple tuple) throws IOException {

		DataBag bag = (DataBag) tuple.get(1);

		Iterator<Tuple> iterator = bag.iterator();
		System.out.println("size of bag:" + bag.size());
		while (iterator.hasNext()) {

			Tuple tup = iterator.next();
			
			String hyphen = new String((String)tup.get(_h));
		

			//Order hyphen tokens
			
			String t[] = hyphen.split("\t");
			
			List<String> order = Arrays.asList(t);
			Collections.sort(order);
			StringBuffer  sorted_= new StringBuffer();
			
			for(int i =0; i<order.size();i++){
				sorted_.append(order.get(i) + "_");
				//sorted = sorted +  order.get(i) + "_";
				
			}
			
			String sorted = new String(sorted_.toString());
			t=null;
			sorted_=null;
			
			if(sorted.endsWith("_")){
				
				sorted = new String(sorted.substring(0, sorted.length()-1));
			}
			
			//order.clear();
			order=null;
			LinkedList<String> originals = new LinkedList<String>();
		
			if(ordered.containsKey(sorted)){
				
				originals =ordered.get(sorted);
				
				if(!originals.contains(hyphen)){
					originals.add(hyphen);
				}
				
				ordered.put(sorted, originals);
				
			}else{
				originals.add(hyphen);
				ordered.put(sorted, originals);
				
			}
			
			hyphen=null;
		
		}
		
		initialized = true;
	}

	
	
	public DataBag exec(Tuple input) throws IOException {
			if (!initialized)
				init(input);

			
			DataBag bag = (DataBag)input.get(0);

			
			//reading tags bookmarks
			Iterator<Tuple> iterator = bag.iterator();
			LinkedList<String> tags= new LinkedList<String>();
			while (iterator.hasNext()) {

				Tuple tup = iterator.next();
				
				if(tup.size()>_tag && tup.get(_tag)!=null){
		
				String tag = (String) tup.get(_tag);
				tags.add(tag);
				}
			}
			
			//create permutations 
			DataBag output = bagFactory.newDefaultBag();
			
			if(tags.size()<2){
				
				return output;
			}
			
			
			HashSet<String> permuta = Permutations.permutations(tags);
			
			//check if any of the permutations is in the dictionary 
			
			
			HashSet<String> hyphens_founds = new HashSet<String>();
			
			
			Iterator<String> iter = permuta.iterator();
			
			while(iter.hasNext()){
				
				String perm = iter.next();
				
				if(ordered.containsKey(perm)){
					
					hyphens_founds.addAll(ordered.get(perm));
					
				}
			}
			
			
			
			
			Iterator<String> iter2= hyphens_founds.iterator();
			while(iter2.hasNext()){
				
				String temp = iter2.next();
				
				Tuple t = TupleFactory.getInstance().newTuple();
				t.append(temp);	
				t.append(1);
				output.add(t);
			}

			return output;

	
	}	
	
	public static void main(String args[]){
		
		LinkedList<String> tags = new LinkedList<String>();
		
		tags.add("e");
		tags.add("c");
		tags.add("b");
		tags.add("a");
		tags.add("d");
		
		MultiWordFeatureGenerator gem = new MultiWordFeatureGenerator();
		HashSet<String> perm = Permutations.permutations(tags);
		LinkedList<String> n = new LinkedList<String>();
		
		
		Iterator<String> iter = perm.iterator();
		
		while(iter.hasNext()){
			
			System.out.println(iter.next());
		}
		
	}
		

	
}